package edu.wisc.hevea.messaging;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Executors;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;

import com.google.protobuf.RpcController;
import com.google.protobuf.ServiceException;
import com.googlecode.protobuf.pro.duplex.CleanShutdownHandler;
import com.googlecode.protobuf.pro.duplex.PeerInfo;
import com.googlecode.protobuf.pro.duplex.RpcClientChannel;
import com.googlecode.protobuf.pro.duplex.RpcConnectionEventNotifier;
import com.googlecode.protobuf.pro.duplex.client.DuplexTcpClientBootstrap;
import com.googlecode.protobuf.pro.duplex.execute.ThreadPoolCallExecutor;
import com.googlecode.protobuf.pro.duplex.listener.RpcConnectionEventListener;
import com.googlecode.protobuf.pro.duplex.logging.CategoryPerServiceLogger;

import edu.wisc.hevea.messaging.Messaging.MessagingService.BlockingInterface;

import edu.wisc.hevea.messaging.Messaging.PutMessageRequest;
import edu.wisc.hevea.messaging.Messaging.PutMessageResponse;
import edu.wisc.hevea.messaging.Messaging.GetMessageRequest;
import edu.wisc.hevea.messaging.Messaging.GetMessageResponse;

public class MessagingServiceClient {

	private static Log log = LogFactory.getLog(MessagingServiceClient.class);
	private static MessagingServiceClient _messagingServiceClient;

	private String clientHostname;
	private int clientPort;

	private Map<String, BlockingInterface> messagingServiceInterfaces;
	private Map<String, RpcController> controllers;

	private MessagingServiceClient(String hostName) {
		String[] sub = hostName.split(":");
		clientHostname = sub[0];
		clientPort = Integer.parseInt(sub[1]);
		messagingServiceInterfaces = new HashMap<String, BlockingInterface>();
		controllers = new HashMap<String, RpcController>();
	}
	
	public synchronized void removeServiceInterface(String brokerServerName) {
		messagingServiceInterfaces.remove(brokerServerName);
		controllers.remove(brokerServerName);
	}

	private synchronized void initializeClient(String brokerServerName) {

		String[] sub = brokerServerName.split(":");
		String serverHostname = sub[0];
		int serverPort = Integer.parseInt(sub[1]);

		PeerInfo client = new PeerInfo(clientHostname, clientPort);
		PeerInfo server = new PeerInfo(serverHostname, serverPort);

		CleanShutdownHandler shutdownHandler = new CleanShutdownHandler();
		try {
			DuplexTcpClientBootstrap bootstrap = new DuplexTcpClientBootstrap(
					client, new NioClientSocketChannelFactory(
							Executors.newCachedThreadPool(),
							Executors.newCachedThreadPool()),
					new ThreadPoolCallExecutor(3, 10));

			// bootstrap.setCompression(compress);

			// RPC payloads are uncompressed when logged - so reduce logging
			CategoryPerServiceLogger logger = new CategoryPerServiceLogger();
			logger.setLogRequestProto(false);
			logger.setLogResponseProto(false);
			bootstrap.setRpcLogger(logger);

			shutdownHandler.addResource(bootstrap);

			// Set up the event pipeline factory.
			// setup a RPC event listener - it just logs what happens
			RpcConnectionEventNotifier rpcEventNotifier = new RpcConnectionEventNotifier();
			RpcConnectionEventListener listener = new RpcConnectionEventListener() {

				@Override
				public void connectionReestablished(
						RpcClientChannel clientChannel) {
					log.info("connectionReestablished " + clientChannel);
				}

				@Override
				public void connectionOpened(RpcClientChannel clientChannel) {
					log.info("connectionOpened " + clientChannel);
				}

				@Override
				public void connectionLost(RpcClientChannel clientChannel) {
					log.info("connectionLost " + clientChannel);
				}

				@Override
				public void connectionChanged(RpcClientChannel clientChannel) {
					log.info("connectionChanged " + clientChannel);
				}
			};
			rpcEventNotifier.setEventListener(listener);
			bootstrap.registerConnectionEventListener(rpcEventNotifier);

			RpcClientChannel channel = bootstrap.peerWith(server);

			BlockingInterface messagingServiceInterface = Messaging.MessagingService
					.newBlockingStub(channel);
			RpcController controller = channel.newRpcController();

			messagingServiceInterfaces.put(brokerServerName,
					messagingServiceInterface);
			controllers.put(brokerServerName, controller);

		} catch (Exception e) {
			e.printStackTrace();
		}

	}

	public static synchronized MessagingServiceClient getMessagingServiceClient(
			String myHostName) {
		if (_messagingServiceClient == null) {
			_messagingServiceClient = new MessagingServiceClient(myHostName);
		}
		return _messagingServiceClient;
	}

	public String sendMessage(String topic, String content,
			String brokerServerName) throws ServiceException {

		if (!messagingServiceInterfaces.containsKey(brokerServerName)) {
			initializeClient(brokerServerName);
		}

		BlockingInterface messagingServiceInterface = messagingServiceInterfaces
				.get(brokerServerName);
		RpcController controller = controllers.get(brokerServerName);

		PutMessageRequest request = PutMessageRequest.newBuilder()
				.setTopic(topic).setContent(content).build();

		PutMessageResponse response = messagingServiceInterface.putMessage(
				controller, request);
		return response.getMsgId();
	}
}
